package wsz.kafka.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.Optional;

/**
 * kafka主题监听
 * @author wsz
 * @date 2022/3/17 20:52
 **/
@Configuration
public class MyKafkaListener {

    @Bean
    public NewTopic topic1() {
        return new NewTopic("topic_1", 1, (short) 1);
    }

    @KafkaListener(topics = "topic_1")
    public void onMessage(ConsumerRecord<Integer, String> record) {
        Optional<ConsumerRecord<Integer, String>> optional = Optional.ofNullable(record);
        if (optional.isPresent()) {
            System.out.println(
                    record.topic() + "\t"
                            + record.partition() + "\t"
                            + record.offset() + "\t"
                            + record.key() + "\t"
                            + record.value());
        }
    }
}
